package io.xxx.omni.osc

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.TypeReference
import io.xxx.omni.osc.common.dateTimeFormatPattern
import io.xxx.omni.osc.domain.Document
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.Serdes
import org.apache.kafka.common.utils.Bytes
import org.apache.kafka.streams.KafkaStreams
import org.apache.kafka.streams.StreamsBuilder
import org.apache.kafka.streams.StreamsConfig
import org.apache.kafka.streams.kstream.Materialized
import org.apache.kafka.streams.kstream.TimeWindowedDeserializer
import org.apache.kafka.streams.kstream.TimeWindowedSerializer
import org.apache.kafka.streams.kstream.TimeWindows
import org.apache.kafka.streams.state.WindowStore
import java.time.Duration
import java.util.*


fun main() {
    val properties = Properties()
    properties[StreamsConfig.APPLICATION_ID_CONFIG] = "demo1"
    properties[StreamsConfig.BOOTSTRAP_SERVERS_CONFIG] = "192.168.206.91:9092,192.168.206.92:9092,192.168.206.93:9092"
    properties[StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG] = Serdes.String().javaClass
    properties[StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG] = Serdes.String().javaClass

    properties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
    properties[StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG] = 0

    val windowedSerializer = TimeWindowedSerializer(Serdes.String().serializer())
    val windowedDeserializer = TimeWindowedDeserializer(Serdes.String().deserializer())
    val windowedSerde = Serdes.serdeFrom(windowedSerializer, windowedDeserializer)

    val typeReference = object : TypeReference<MutableList<Document>>() {}
    val builder = StreamsBuilder()
    builder.stream<String, String>("DOCUMENT")
        .selectKey { _, value ->
            val document = JSON.parseObject(value, Document::class.java)
            document.storeId
        }
        .groupByKey()
        .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
        .aggregate(
            { "" },
            { _, value, aggregate ->
                val document = JSON.parseObject(value, Document::class.java)
                if (aggregate == "" || aggregate == " --- ") {
                    JSON.toJSONStringWithDateFormat(listOf(document), dateTimeFormatPattern)
                } else {
                    val list = JSON.parseObject(aggregate, typeReference)
                    list.add(document)
                    JSON.toJSONStringWithDateFormat(list, dateTimeFormatPattern)
                }
            },
            Materialized.`as`<String, String, WindowStore<Bytes, ByteArray>>("document-snapshots")
                .withKeySerde(Serdes.String())
        )
        .toStream()
        .to("stream-test")
//        .to("stream-test", Produced.with(windowedSerde, Serdes.String()))

    val streams = KafkaStreams(builder.build(), properties)
    streams.start()
}